package com.jy.oms.framework.asyn.task.cluster.config;

import com.alibaba.nacos.api.NacosFactory;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.Listener;
import com.jy.oms.framework.asyn.task.core.ExecutorConfig;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.exception.TaskConfigException;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerProcessor;
import com.jy.oms.framework.asyn.task.util.AssertTask;
import com.jy.oms.framework.asyn.task.util.ServerConfigAdapter;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;

import java.io.StringReader;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;

/**
 * nacos配置服务
 *
 * @author liuquan
 */
@Service
@Primary
public class NacosTaskConfigService implements ITaskConfigService, ITaskHandlerProcessor {
    private static Logger logger = LoggerFactory.getLogger(NacosTaskConfigService.class);

    private static String configGroup = "asyn-task";
    private ConfigService configService;
    private Map<String, String> configMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, ITaskConfigChangeListener> listenerMap = new ConcurrentHashMap<>();

    @Override
    public int getProcessorOrder() {
        return ITaskHandlerProcessor.PROCESSOR_LOWEST_ORDER;
    }

    @Override
    public void processAfterTaskHandlerInitialization() {
        this.init();
    }

    @Override
    public void init() {
        try {
            ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
            if (executorConfig == null || !executorConfig.isEnableTaskExecutor()) {
                return;
            }

            Properties properties = ServerConfigAdapter.getNacosProperties();
            configService = NacosFactory.createConfigService(properties);

            //读取已配置数据
            String config = configService.getConfig(executorConfig.getBusinessGroup(), configGroup, 3000);
            if (StringUtils.isNotBlank(config)) {
                configMap.putAll(this.parseConfig(config));
            }

            //配置加载完后触发配置变更事件
            if(!configMap.isEmpty()){
                for (Map.Entry<String, String> entry : configMap.entrySet()) {
                    ITaskConfigChangeListener listener = listenerMap.get(entry.getKey());

                    if (listener == null) {
                        continue;
                    }

                    try {
                        listener.onConfigChangeAfterLoaded(entry.getKey(), entry.getValue());
                    } catch (Exception e) {
                        logger.error("监听事件执行失败，configKey:{},configValue:{}", entry.getKey(), entry.getValue(), e);
                    }
                }
            }

            //监听配置变化
            configService.addListener(executorConfig.getBusinessGroup(), configGroup, new Listener() {
                @Override
                public void receiveConfigInfo(String config) {
                    logger.info("任务配置发生变化, 【{}】", config);

                    Map<String, String> map = parseConfig(config);
                    List<String> changeConfigKeys = new ArrayList<>();
                    for (String configKey : listenerMap.keySet()) {
                        String newConfigValue = map.get(configKey);
                        String oldConfigValue = configMap.get(configKey);
                        if (!StringUtils.equals(newConfigValue, oldConfigValue)) {
                            changeConfigKeys.add(configKey);
                        }
                    }
                    configMap.clear();
                    configMap.putAll(map);

                    for (String configKey : changeConfigKeys) {
                        ITaskConfigChangeListener listener = listenerMap.get(configKey);

                        if (listener == null) {
                            continue;
                        }

                        try {
                            listener.onConfigChange(configKey, configMap.get(configKey));
                        } catch (Exception e) {
                            logger.error("监听事件执行失败，configKey:{},configValue:{}", configKey, configMap.get(configKey), e);
                        }
                    }
                }

                @Override
                public Executor getExecutor() {
                    return null;
                }
            });
        } catch (Exception e) {
            throw new TaskConfigException("nacos配置错误", e);
        }
    }

    private Map<String, String> parseConfig(String config) {
        Map<String, String> configMap = new HashMap<>();
        if (StringUtils.isBlank(config)) {
            return configMap;
        }

        try {
            Properties prop = new Properties();
            prop.load(new StringReader(config));

            for (Map.Entry<Object, Object> entry : prop.entrySet()) {
                String key = StringUtils.trimToEmpty(entry.getKey().toString());
                if (key.length() == 0) {
                    continue;
                }
                String value = StringUtils.trimToEmpty(entry.getValue().toString());
                configMap.put(key, value);
            }
            return configMap;
        } catch (Exception e) {
            logger.error("配置解析报错", e);
        }

        return configMap;
    }

    @Override
    public String get(String key) {
        return configMap.get(key);
    }

    @Override
    public void addTaskConfigChangeListener(ITaskConfigChangeListener listener) {
        AssertTask.notNull(listener, "listener不能为空");
        AssertTask.notNull(listener.getConfigKey(), "listener key不能为空");
        ITaskConfigChangeListener oldListener = listenerMap.putIfAbsent(listener.getConfigKey(), listener);
        AssertTask.isNull(oldListener, "相同key不能存在多个监听器");
    }

}
